package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/2/5
 */
public class Demo4_Agg
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());
        Table table = tableEnvironment.fromDataStream(ds);

        table.printSchema();


        //统计每种传感器的vc之和
        Table result = table
                            .groupBy($("id"))
                            .select($("id"),  $("vc").sum().as("sumVC"));


        /*
            Exception in thread "main" org.apache.flink.table.api.TableException:
                Table sink 'default_catalog.default_database.Unregistered_DataStream_Sink_1'
                doesn't support consuming update changes which is produced by node
                 GroupAggregate(groupBy=[id], select=[id, SUM(vc) AS EXPR$0])

                 写出的表不支持消费更新的变化。
                    表转换为流分为三种。
                        Append-Only流：  表中的数据只会追加(新增)  toDataStream()

                    表中的数据会发生更新或变化(聚合):
                            upsert(toChangelogStream) 流：  适合变化的场景  toChangelogStream
                                    显示出流中数据变化的过程。
                                        +I: 新增一条数据
                                        -U: 更新前
                                        +U： 更新后
                            retract 流:  适合变化的场景。 对比ChangelogStream，多了一个标记。  toRetractStream

         */
        //DataStream<Row> ds2 = tableEnvironment.toChangelogStream(result);

        /*
                第一位是 Boolean，代表当前数据是否是更新后的数据，标记
                第二位是 Row,真正的数据
         */
        DataStream<Tuple2<Boolean, Row>> ds2 = tableEnvironment.toRetractStream(result, Row.class);

        ds2.filter(t -> t.f0)
           .map(t -> t.f1)
            .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
